When the API server serves logs for a RUNNING task instance, FileTaskHandler constructs the configured executor to call get_task_log() and never starts or shuts it down. KubernetesExecutor.__init__ eagerly created a multiprocessing.Manager(), which forks a serve_forever child process. Because the executor instance is cached per process, this leaks one orphaned Manager process (~350-400 MB resident) per API-server worker. With gunicorn/uvicorn worker recycling, each refreshed worker's Manager reparents to PID 1 and is never reaped, so they accumulate and push the API server toward OOM.

get_task_log() only needs the kube client and the pod namespace; it never touches task_queue/result_queue/_manager. The Manager is purely a scheduling-loop resource.

Closes #68693.

Fix

Create the multiprocessing.Manager() and its queues lazily in start() instead of __init__(). The scheduling loop (start/execute_async/sync/end) is their only consumer, and the scheduler always calls start() first. end() now no-ops when the executor was never started. Constructing the executor purely to read logs no longer spawns a Manager.

Design rationale

• Why start() and not a lazy property? The queues are only used after the scheduler calls start(); nothing touches them between construction and start(). Tying creation to start() keeps the lifecycle explicit and pairs cleanly with end().
• Consistent with LocalExecutor. LocalExecutor already defers its queue/process creation to start(), so it never leaked on this path. This change brings KubernetesExecutor in line with that pattern, which also covers LocalKubernetesExecutor (its embedded KubernetesExecutor is constructed in __init__ and only started via LocalKubernetesExecutor.start()).

Affected path

• FileTaskHandler._read() calls the executor's get_task_log for running tasks: file_task_handler.py#L637
• which loads/instantiates the default executor: executor_loader.py#L292

Backward compatibility

No public API change. task_queue/result_queue/_manager are None until start() is called; all internal consumers run after start(). Behavior for a started executor (the scheduler) is unchanged.